package com.shujia.flink.window

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{EventTimeSessionWindows, ProcessingTimeSessionWindows}
import org.apache.flink.streaming.api.windowing.time.Time

object Demo2SessionWindow {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    //设置事件模式
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val ds1: DataStream[String] = env.socketTextStream("master", 6666)

    val kvDS: DataStream[(String, Int)] = ds1
      .flatMap(_.split(","))
      .map((_, 1))

    /**
      * ProcessingTimeSessionWindows  处理时间的会话窗口
      *
      * EventTimeSessionWindows  事件事件的会话窗口
      *
      */

    /* kvDS
       .keyBy(_._1)
       .window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)))
       .sum(1)
       .print()*/
    /*
001,1599810953000
001,1599810954000
001,1599810955000
001,1599810957000
001,1599810956000
001,1599810961000
001,1599810965000
001,1599810966000
001,1599810966000
001,1599810966000
001,1599810966000
001,1599810966000
001,1599810970000
001,1599810980000
002,1599810955000
      */

    ds1.map(line => (line.split(",")(0), line.split(",")(1).toLong))
      //指定时间列
      .assignAscendingTimestamps(_._2)
      .keyBy(_._1)
      //事件事件窗口
      .window(EventTimeSessionWindows.withGap(Time.seconds(5)))
      .sum(1)
      .print()


    env.execute()
  }
}
